package com.zidongxiangxi.rabbit.producer.job;

import com.zidongxiangxi.rabbit.core.job.AbstractWorkThread;
import com.zidongxiangxi.rabbit.producer.config.ProducerSequenceClearConfig;
import com.zidongxiangxi.rabbit.producer.entity.ProducerSequence;
import com.zidongxiangxi.rabbit.producer.manager.ProducerSequenceManager;
import org.springframework.util.CollectionUtils;

import java.util.Date;
import java.util.List;
import java.util.stream.Collectors;

/**
 * RabbitMQ顺序消息记录清理线程
 *
 * @author chenxudong
 * @date 2020/11/07
 */
public class ProducerSequenceClearThread extends AbstractWorkThread {
	private final long retentionPeriod;
	private final int clearBatchSize;
	private final long workPeriod;
	private final ProducerSequenceManager producerSequenceManager;

	public ProducerSequenceClearThread(ProducerSequenceClearConfig config, ProducerSequenceManager producerSequenceManager) {
		this.retentionPeriod = config.getRetentionPeriod() * 86400 * 1000;
		this.clearBatchSize = config.getBatchSize();
		this.workPeriod = config.getWorkPeriod();
		this.producerSequenceManager = producerSequenceManager;
	}

	@Override
	public void work() {
		if (retentionPeriod <= 0) {
			stop();
			return;
		}
		Date beforeTime = new Date(System.currentTimeMillis() - retentionPeriod);
		int size = clearRecord(beforeTime, clearBatchSize);
		if (size < clearBatchSize) {
			try {
				Thread.sleep(workPeriod);
			} catch (InterruptedException ignore) {}
		}
	}

	private int clearRecord(Date beforeTime, int batchSize) {
		List<ProducerSequence> sequenceMessages = producerSequenceManager.list(beforeTime, batchSize);
		if (CollectionUtils.isEmpty(sequenceMessages)) {
			return 0;
		}
		List<Long> ids = sequenceMessages.stream().map(ProducerSequence::getId).collect(Collectors.toList());
		producerSequenceManager.deleteByIds(ids);
		return ids.size();
	}

}
